package com.atguigu.app.dws;

import com.atguigu.bean.ProvinceStats;
import com.atguigu.uitls.ClickHouseUtil;
import com.atguigu.uitls.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//数据流：web/app -> nginx -> 业务服务器 -> Mysql(binlog) -> FlinkApp -> Kafka(ODS) -> FlinkApp -> Kafka/HBase(DWD/DIM) -> FlinkApp(Redis) -> Kafka(DWM) -> FlinkApp -> ClickHouse
//程  序：Mock -> Mysql(binlog) -> FlinkCDC -> Kafka(ZK) ->  BaseDBApp -> Kafka/HBase(ZK,HDFS) -> OrderWideApp(Redis) -> Kafka(ZK) -> ProvinceStatsSqlApp -> ClickHouse
public class ProvinceStatsSqlApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境与Kafka主题的分区数保持一致

        //CK
        //        env.setStateBackend(new FsStateBackend("hdfs://"));
        //        env.enableCheckpointing(5000L);
        //        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        //        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //        env.getCheckpointConfig().setCheckpointInterval(10000L);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.使用DDL方式读取Kafka主题的数据创建表,提取事件时间生成WaterMark
        String groupId = "province_stats_210826";
        String orderWideTopic = "dwm_order_wide";

        tableEnv.executeSql("create table order_wide( " +
                "    order_id bigint, " +
                "    province_id bigint, " +
                "    split_total_amount DECIMAL, " +
                "    province_name String, " +
                "    province_area_code String, " +
                "    province_iso_code String, " +
                "    province_3166_2_code String, " +
                "    create_time String, " +
                "    rt as TO_TIMESTAMP(create_time), " +
                "    WATERMARK FOR rt AS rt - INTERVAL '2' SECOND " +
                ") with (" + MyKafkaUtil.getKafkaDDL(orderWideTopic, groupId) + ")");

        //TODO 3.开窗、计算结果
        Table resultTable = tableEnv.sqlQuery("" +
                "select " +
                "    DATE_FORMAT(TUMBLE_START(rt, INTERVAL '10' second),'yyyy-MM-dd HH:mm:ss') stt, " +
                "    DATE_FORMAT(TUMBLE_end(rt, INTERVAL '10' second),'yyyy-MM-dd HH:mm:ss') edt, " +
                "    province_id, " +
                "    province_name, " +
                "    province_area_code, " +
                "    province_iso_code, " +
                "    province_3166_2_code, " +
                "    count(distinct order_id) order_count, " +
                "    sum(split_total_amount) order_amount, " +
                "    UNIX_TIMESTAMP()*1000 ts " +
                "from order_wide " +
                "group by " +
                "    province_id, " +
                "    province_name, " +
                "    province_area_code, " +
                "    province_iso_code, " +
                "    province_3166_2_code, " +
                "    TUMBLE(rt, INTERVAL '10' second)");

        //TODO 4.将动态表转换为流
        DataStream<ProvinceStats> provinceStatsDataStream = tableEnv.toAppendStream(resultTable, ProvinceStats.class);

        //TODO 5.将数据写出到ClickHouse
        provinceStatsDataStream.print(">>>>>>>>>>>");
        provinceStatsDataStream.addSink(ClickHouseUtil.getSinkFunction("insert into province_stats_210826 values(?,?,?,?,?,?,?,?,?,?)"));

        //TODO 6.启动任务
        env.execute("ProvinceStatsSqlApp");
    }

}
